<?php
/*
 * 添加异步任务队列功能类,适用于批量业务，异步业务
 * @author HelloWorld
 * @version 2021/12/2
 * */
class Module_Queue_Base{
    //异步业务队列list key
    public static $list_key = 'Module_Queue_Asychronous';
    public static $repair_key = 'Module_Queue_Repair_Asychronous';

    /* 入队列方法
     * @param task array
     * pe 为业务类型，对应处理业务的类名，命名规范为Module_Queue_Base[pe_name]
     * method 为业务处理单元名称，对应业务类方法名Module_Queue_Base[pe_name]::[method_name]
     * 数组格式要求 1维的key必须包括 pe method data
     * */
    public static function addQueue($task,$fast = 0,$key = 1){
        if($key == 1){
            $list_key = self::$list_key;
        }elseif ($key == 2){
            $list_key = self::$repair_key;
        }else{
            $list_key = self::$list_key;
        }

        if(empty($task) || !is_array($task)){
            Func_Log::write('Queue','Error:invalid_params');
            return false;
        }
        if(!isset($task['data']) || !is_array($task['data']) || !isset($task['type']) || !isset($task['method'])){
            Func_Log::write('Queue','Error:miss_params:'.json_encode($task));
            return false;
        }
        $task['data']['task_id'] = uniqid();
        Func_Log::write('Queue','addQueue','task_id_'.$task['data']['task_id'].',method='.$task['method']);
        if($fast){
            $res = Lib_Redis_List::rPush($list_key,serialize($task));
        }else{
            $res = Lib_Redis_List::lPush($list_key,serialize($task));
        }
        if(!$res){
            Func_Log::write('Queue',$task['data']['task_id'],'Error:add_queue_failed:'.json_encode($task));
        }

        return $res;
    }

    /*
     * 批量添加
     * data 部分相对于单任务 为二维数组
     * */
    public static function addQueueBat($tasks){
         Func_Log::write('Queue','addQueueBat','task_id_'.$tasks['task_id']);
        if(empty($tasks) || !is_array($tasks)){
             Func_Log::write('Queue',$tasks['task_id'],'Error:invalid_params:'.$tasks);
            return false;
        }
        if(!isset($tasks['data']) || !is_array($tasks['data']) || !isset($tasks['type']) || !isset($tasks['method'])){
             Func_Log::write('Queue',$tasks['task_id'],'Error:miss_params:'.json_encode($tasks));
            return false;
        }
        //子任务标识计数器
        $i = 1;
        foreach($tasks['data'] as $task){
            $queue['type'] = $tasks['type'];
            $queue['method'] = $tasks['method'];
            $queue['data'] = $task;
            $queue['data']['task_id'] = $tasks['task_id'].'_'.$i;
            $i++;
            $res = Lib_Redis_List::lPush(self::$list_key,serialize($queue));
            if(!$res){
                 Func_Log::write('Queue',$queue['data']['task_id'],'Error:add_queue_failed:'.json_encode($queue));
            }
        }
        return true;
    }
}

?>
